package com.ict.dataprocess;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

import org.apache.log4j.Logger;

import com.google.protobuf.InvalidProtocolBufferException;
import com.ict.config.Config;
import com.ict.utils.CacheObject;
import com.ict.utils.ConnMQUtil;
import com.ict.utils.ConnRedisUtil;
import com.rabbitmq.client.ConsumerCancelledException;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.ShutdownSignalException;

import team.iOceanPlus.PB.AISData.PBAISDynamic;
import team.iOceanPlus.PB.CommonEnum.PB_Enum_TargetInfo_Type;
import team.iOceanPlus.PB.Target;
import team.iOceanPlus.PB.Target.PBTargetPosition;

public class FuseMQToRedis extends Thread {
	private static  Logger  logger=Logger.getLogger(FuseMQToRedis.class);
	public void run() {
		List<byte []>  lstCache=CacheObject.returnLstCache();
		if (lstCache.size()==Config.listCatchSize) {
		List<Target.PBTargetPosition> lstRecvPBTargetPos = new ArrayList<Target.PBTargetPosition>();
		
		/*List<byte []>  lstRecvFusePb= ConnRedisUtil.jedis.lrange("recv_fuse_pb".getBytes(), 0, -1);
		ConnRedisUtil.jedis.del("recv_fuse_pb");*/
		for (byte[]  byteArray : lstCache) {
			try {
				lstRecvPBTargetPos.addAll(Target.PBTarget.parseFrom(byteArray).getListTargetPosList());
			} catch (InvalidProtocolBufferException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	
		logger.info("fuse--position  count"+lstRecvPBTargetPos.size());
		try {
			for (int x = 0; x < lstRecvPBTargetPos.size(); x++) {
				Target.PBTargetPosition pos = lstRecvPBTargetPos.get(x);
				if (pos.getEnumTargetInfoType().getNumber() == PB_Enum_TargetInfo_Type.EV_TargetInfoType_AISStatic_VALUE) {
					lstRecvPBTargetPos.remove(x);
					x=x-1;
				}
				 
			}
			/** 去掉重复数据 */
			lstRecvPBTargetPos.stream().distinct().collect(Collectors.toList());
			/** 同一目标位置信息保存最新时间一条 */
			/*List<Target.PBTargetPosition> group = new ArrayList<Target.PBTargetPosition>();*/
			
			for(Target.PBTargetPosition  pos :lstRecvPBTargetPos) {
				String fieldKey=pos.getTargetID()+"";
				if(ConnRedisUtil.jedis.hexists("mid_fuse_pos_new", fieldKey.toString())) {
					byte [] bytRedisPos= ConnRedisUtil.jedis.hget("mid_fuse_pos_new".getBytes(), fieldKey.toString().getBytes());
					Target.PBTargetPosition  redisPos=(Target.PBTargetPosition)byteArrayToObject(bytRedisPos);
						if(pos.getLastAssocUsingDSCombinationUTCTime()>redisPos.getLastAssocUsingDSCombinationUTCTime()) {
							ConnRedisUtil.jedis.hset("mid__fuse_pos_new".getBytes(), fieldKey.toString().getBytes(), objectToByteArray(pos));
						}
					
				}else {
					ConnRedisUtil.jedis.hset("mid_fuse_pos_new".getBytes(), fieldKey.toString().getBytes(), objectToByteArray(pos));
				}
				
			}
			/*Map<byte[] ,byte[]> newMap= ConnRedisUtil.jedis.hgetAll("mid_fuse_pos_new".getBytes());
			Map<byte[] ,byte[]> oldMap= ConnRedisUtil.jedis.hgetAll("mid_fuse_pos_old".getBytes());*/
			/** 目标点没有变化的移除 */
			/*for(byte[] bytPos :newMap.values()) {
				group.add((Target.PBTargetPosition) byteArrayToObject(bytPos));
			}*/
			/*ConnRedisUtil.jedis.rename("mid_fuse_pos_new", "send_fuse_pb");*/
			/*logger.info("fuse--position  filted cnt is"+group.size());*/
			
			/*for(  int  x=0;x<group.size() ;x++) {
				ConnRedisUtil.jedis.hset("send_fuse_pb".getBytes(), ("field"+x).getBytes(), objectToByteArray(group.get(x)));
			}
			*/
			/*Target.PBTarget.Builder targetBuilder = Target.PBTarget.newBuilder();
			targetBuilder.setEnumSenderSoftwareValue(6);
			targetBuilder.addAllListTargetPos(group);
			Target.PBTarget pbTarget = targetBuilder.build();
			ConnRedisUtil.jedis.hset("send_orig_pb".getBytes(), ("field").getBytes(), pbTarget.toByteArray());*/
			/** 融合模式 */
			/*
			 * String[] arrRoutingKey2 = { "OnLine.FusedData.TargetTracks" }; ConnMQUtil
			 * connMQUtil2 = new ConnMQUtil(); QueueingConsumer consumer2;
			 * 
			 * consumer = connMQUtil.recv("topic", "xgs", arrRoutingKey2, "xgs_tap_by_tlc");
			 * QueueingConsumer.Delivery delivery2 = consumer.nextDelivery();
			 * Target.PBTarget info2 = Target.PBTarget.parseFrom(delivery2.getBody());
			 *//** 去掉AIS静态数据 */
			/*
			 * for (int x = 0; x < info2.getListTargetPosList().size(); x++) {
			 * Target.PBTargetPosition pos= info2.getListTargetPosList().get(x);
			 * if(pos.getEnumTargetInfoType().getNumber()==3) {
			 * info2.getListTargetPosList().remove(x);
			 * 
			 * 
			 * } }
			 *//** 去掉重复数据 */
			/*
			 * info2.getListTargetPosList().stream().distinct().collect(Collectors.toList())
			 * ;
			 *//** 同一目标位置信息保存最新时间一条 */
			/*
			 * List<Target.PBTargetPosition> group2 = (List<PBTargetPosition>)
			 * info2.getListTargetPosList().stream().limit(1).collect(Collectors.groupingBy(
			 * t -> t.getTargetID()));
			 *//** 目标点没有变化的移除 *//*
								 * for (int x = 0; x < group.size(); x++) { Target.PBTargetPosition pos=
								 * info2.getListTargetPosList().get(x); Boolean
								 * isContainsPos=CacheObject.lstOrigTargetCache.get(0).getListTargetPosList().
								 * contains(pos); if(isContainsPos) { group2.remove(group.get(x)); } }
								 * CacheObject.lstTargetCache.clear(); CacheObject.lstTargetCache.add(info2);
								 * System.out.println("recving redis");
								 * ConnMQUtil.recvChannel.basicAck(delivery.getEnvelope().getDeliveryTag(),
								 * false);
								 */

		} catch (ShutdownSignalException e1) {
			// TODO Auto-generated catch block
			e1.printStackTrace();
		} catch (ConsumerCancelledException e1) {
			// TODO Auto-generated catch block
			e1.printStackTrace();
		}
		System.out.println("second over");
		FuseRedisToMQ third = new FuseRedisToMQ();
		third.run();
		lstCache.clear();
	}	
		

	}

	
	
	
	
	public static byte[] objectToByteArray(Object obj) {
        byte[] bytes = null;
        ByteArrayOutputStream byteArrayOutputStream = null;
        ObjectOutputStream objectOutputStream = null;
        try {
            byteArrayOutputStream = new ByteArrayOutputStream();
            objectOutputStream = new ObjectOutputStream(byteArrayOutputStream);
            objectOutputStream.writeObject(obj);
            objectOutputStream.flush();
            bytes = byteArrayOutputStream.toByteArray();

        } catch (IOException e) {
        } finally {
            if (objectOutputStream != null) {
                try {
                    objectOutputStream.close();
                } catch (IOException e) {
                }
            }
            if (byteArrayOutputStream != null) {
                try {
                    byteArrayOutputStream.close();
                } catch (IOException e) {
                }
            }

        }
        return bytes;
    }

	
	public static Object byteArrayToObject(byte[] bytes) {
        Object obj = null;
        ByteArrayInputStream byteArrayInputStream = null;
        ObjectInputStream objectInputStream = null;
        try {
            byteArrayInputStream = new ByteArrayInputStream(bytes);
            objectInputStream = new ObjectInputStream(byteArrayInputStream);
            obj = objectInputStream.readObject();
        } catch (Exception e) {
        } finally {
            if (byteArrayInputStream != null) {
                try {
                    byteArrayInputStream.close();
                } catch (IOException e) {
                }
            }
            if (objectInputStream != null) {
                try {
                    objectInputStream.close();
                } catch (IOException e) {
                }
            }
        }
        return obj;
    }


}
